package test.mq;

import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;

import hyl.base.mq.rocket.MyRocketMq;
import hyl.core.MyDate;
import hyl.core.info.CMsg;
import hyl.core.run.IFinishT;

public class testrocket_send {
	MyRocketMq mqp = MyRocketMq.iniProducer("producer_group1", "group2");

	public static void main(String[] args) throws Exception {
		testrocket_send t = new testrocket_send();
		t.test();
	}

	public void test() throws Exception {
		mqp.startProducer();
		// pub1() ;
		// pub2();
		// pub3();
		pub6();
		// pub4() ;

		//

	}

	IFinishT<CMsg> if2 = new IFinishT<CMsg>() {

		@Override
		public boolean success(CMsg o) {
			System.out.println("发送成功" + o);
			return false;
		}

		@Override
		public void error(CMsg o, Throwable e) {
			// TODO Auto-generated method stub

		}

	};

	public void pub6() throws Exception {

		mqp.set主题("test2").set回调函数(if2);
		System.out.println("开始发送....");
		for (int i = 0; i < 5; i++) {

			mqp.send("ke键_" + i, ("va值=" + i).getBytes());

			MyDate.sleep(200);
		}
		System.out.println("发送完毕....");

	}

	public void pub5() throws Exception {

		System.out.println("开始发送....");
		for (int i = 0; i < 100; i++) {
			Message msg = mqp.createMsg("test", "ke键_" + i, "va值=" + i);
			mqp.sendOne(msg);
			MyDate.sleep(500);
		}
		System.out.println("发送完毕....");

	}

	public void pub4() throws Exception {

		System.out.println("开始发送....");
		Message msg = mqp.createMsg("test", "ke键41", "va值1");
		mqp.sendWithID(1, msg, null);
		msg = mqp.createMsg("test", "ke键42", "va值2");
		mqp.sendWithID(2, msg, null);
		msg = mqp.createMsg("test", "ke键43", "va值3");
		mqp.sendWithID(3, msg, null);
	}

	public void pub3() throws Exception {

		System.out.println("开始发送....");
		Message msg = mqp.createMsg("test", "ke键31", "va值1");
		mqp.sendWithCall(msg, scb);
		msg = mqp.createMsg("test", "ke键32", "va值2");
		mqp.sendWithCall(msg, scb);
		msg = mqp.createMsg("test", "ke键33", "va值3");
		mqp.sendWithCall(msg, scb);
	}

	public void pub2() throws Exception {

		System.out.println("开始发送....");
		Message msg = mqp.createMsg("test", "ke键21", "va值1");
		mqp.sendLog(msg);
		msg = mqp.createMsg("test", "ke键22", "va值2");
		mqp.sendLog(msg);
		msg = mqp.createMsg("test", "ke键23", "va值3");
		mqp.sendLog(msg);
	}

	public void pub1() throws Exception {

		System.out.println("开始发送....");
		mqp.set主题("test").set回调函数(if2);

		mqp.send("ke键14", "va值1".getBytes());

		mqp.send("ke键12", "va值2".getBytes());

		mqp.send("ke键13", "va值3".getBytes());
	}

	private SendCallback scb = new SendCallback() {
		@Override
		public void onSuccess(SendResult sendResult) {
			System.out.println("发送成功");
		}

		@Override
		public void onException(Throwable e) {
			System.out.println("发送异常");

		}
	};
}
